package com.wang.kafkastudy.consumer;

import com.wang.kafkastudy.config.KafkaConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class UserLogConsumer {
    private Logger logger = LoggerFactory.getLogger(UserLogConsumer.class);

    @KafkaListener(topics = {KafkaConfig.USER_LOG_TOPIC_NAME})
    public void consumer(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
        logger.info(">>>>>>>>>>>>>>record={}", kafkaMessage);
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            logger.info("消费消息{}", message);
        }
    }
}
